# A simple bootstrap and foreach example adapted from Burns Statistcs,
# http://www.burns-stat.com/pages/Tutor/bootstrap_resampling.html

library('quantmod')
library('foreach')

# Retrieve the daily adjusted close price for Ford and the S&P 500 from
# Yahoo Finance:
F <- getSymbols('F', from='2010-01-01', to='2010-12-31', auto.assign=FALSE)[,6]
GSPC <- getSymbols('^GSPC', from='2010-01-01', to='2010-12-31', auto.assign=FALSE)[,6]

# Convert to returns
F <- diff(log(F))
GSPC <- diff(log(GSPC))

# Compute empirically observed beta
coef(lm(F ~ GSPC))

# Bootstrap to get a sense of variability
n <- length(F)
t1 <- proc.time()
beta <- foreach(j=1:1000, .combine=c,
                .multicombine=TRUE, .inorder=FALSE) %dopar%
{
  ind <- sample(n,n,replace=TRUE)
  coef(lm(F[ind] ~ GSPC[ind]))[2]
}
print(proc.time() - t1)

# Look at the results
hist(beta, col='yellow')
abline(v=coef(lm(F ~ GSPC))[2],col='blue',lwd=2)

# Now let's retry this in parallel. A Redis server is already running locally on
# my netbook. I'll load the doRedis package and register a doRedis job queue
# with foreach.  I'll also start a local worker running on my netbook to
# process jobs.
library('doRedis')
#registerDoRedis(queue='jobs')
registerDoRedis(queue='jobs', host="172.16.0.247")
startLocalWorkers(n=1, queue='jobs')
setChunkSize(250)

# We encapsulate the combine function in a closure that reports
# average performance every time the combine function is called to
# aggregate results.
f <- function()
{
  count <- 0
  x <- 0
  y <- 0
  time <- proc.time()[3]
  function(...)
  {
    count <<- count + length(list(...)) - 1
    dt <- proc.time()[3] - time
    cat("Average iterations  per second: ",count/dt, "\n")
    x <<- c(x, dt)
    y <<- c(y, count/dt)
    plot(x,y,type="l",lwd=2,col=4,xlab="time (s)",
         main="Running average bootstrap iterations/s")
    Sys.sleep(0.01) # Yield to update the plot
    flush.console()
    c(...)
  }
}

g <- f()
# The followingloop is 10x longer than before to help us get
# a feel for the parallel speed up.
beta <- foreach(j=1:10000, .combine=g, .inorder=FALSE,
                .multicombine=TRUE, .maxcombine=250) %dopar%
{
  ind <- sample(n,n,replace=TRUE)
  coef(lm(F[ind] ~ GSPC[ind]))[2]
}

removeQueue("jobs")
